package xubo.wangcaifeng.love2

import com.alibaba.fastjson.JSON
import com.typesafe.config.ConfigFactory
import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.kafka.KafkaCluster.Err
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaCluster, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import xubo.wangcaifeng.love.JedisConnectionPool

object CmccMontitorApp {
  def main(args: Array[String]): Unit = {
    //对配置文件解析的时候需要的一个工具类
    val load = ConfigFactory.load()
    //StreamingContext
    val conf = new SparkConf()
      .setAppName("cmcc_move")
      .setMaster("local[*]")
      //将rdd压缩
      .set("spark.rdd.compress","true")
      //优雅的停止
      .set("spark.streaming.stopGracefullyOnShutdown","true")
      //序列化方式   rdd,   worker和worker数据传输序列化方式
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      // 设置从kafka每个分区每秒钟消费1000条数据
      .set("spark.streaming.kafka.maxRatePerPartition","5000")
    val ssc = new StreamingContext(conf,Seconds(2))
    val sc = ssc.sparkContext
    val pcodeDict = sc.textFile("province/ProvinceRule.txt")
      .map(line => {
        val filed = line.split(",")
        (filed(0), filed(1))
      }).collect().toMap
    //广播字典数据
    val pcodeBC = sc.broadcast(pcodeDict)


    /**
    想要从kafka中实时得拿取数据，SparkStreaming与kafka进行整合了，看官网，有一个工具类
      *
      */
    val groupId = load.getString("group.id")
    val topic = load.getString("kafka.topic")
    //设置kafka的参数  可以将参数写到配置文件中，然后在这边对参数进行解析
    val kafkaParams = Map[String,String](
      "bootstrap.servers" -> load.getString("bootstrap.servers"),
      "auto.offset.reset" -> load.getString("auto.offset.reset"),
      "group.id" -> groupId
    )
    //连接数据库，看有没有偏移量  如果类里面是apply方法，这个可以不写
    val dbOffset: Map[TopicAndPartition, Long] = OffsetUtils(groupId)


    //消费kafka中的数据  四大参数  ssc  kafka参数   偏移量
    //从数据库中查偏移量，如果有则从偏移量开始消费
    val kafkaStream = if (dbOffset.size != 0) {
      //如果不是第一次消费，然后还的校验自己维护的偏移量
      val kafkaCluster = new KafkaCluster(kafkaParams)
      //拿到kafka里面的偏移量
      val offsets: Either[Err, Map[TopicAndPartition, KafkaCluster.LeaderOffset]] = kafkaCluster.getEarliestLeaderOffsets(dbOffset.keySet)
      var checkedOffset = Map[TopicAndPartition,Long]()
      if (offsets.isRight){
        //kafka集群中存储的某个主题某个分区（0,1,2）下最早的有效偏移量
        val partitionToOffset: Map[TopicAndPartition, KafkaCluster.LeaderOffset] = offsets.right.get
        //这是数据库中的偏移量，与partitionToOffset进行对比
        dbOffset.foreach(db=>{
          val clusterTopicAndPartitionOffset: Long = partitionToOffset.get(db._1)//这个是可能有,可能没有
            .get.offset
          if (db._2 < clusterTopicAndPartitionOffset){
            // 数据库中存的便宜量比集中中最早的偏移量要小，说明我们维护的offset过期了
            //拿一个容器放校验后的数据
            checkedOffset += ((db._1,clusterTopicAndPartitionOffset))
          }else{
            checkedOffset += db._1->db._2
          }
        })

      }else{
        checkedOffset=dbOffset
      }
      val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.key(), mmd.message())
      KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, dbOffset, messageHandler)

    } else {
      //如果数据库中没有偏移量则从头开始消费
      //第一次消费数据
      KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, Set(topic))
    }

    //对数据进行处理
    kafkaStream.foreachRDD(rdd =>{
      //将每一个RDD的偏移量插入到数据库中
      //这个偏移量的size与kafka的分区数量是一致的
      val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      //对每一个RDD进行处理
      // 1. 业务概况 充值成功订单量, 充值成功金额, 充值成功率及充值平均时长.
      val baseData: RDD[(String, List[Double] , String)] = rdd.map(_._2)
        .map(JSON.parseObject(_))
        .filter(obj => obj.getString("serviceName").equals("reChargeNotifyReq"))
        .map(obj => {
          val kip = ObjUtils(obj)
          (kip._1, kip._2, kip._3)
        }).cache()
      //查看每天的充值订单，充值金额，充值成功率，充值平均时长
      baseData.map(t => (t._1.substring(0,8),t._2))//按照日期来分
        .reduceByKey((list1,list2) =>{
        list1.zip(list2).map(t=>t._1+t._2)
      }).foreachPartition(
        //it 是RDD中每一个分区的数据，这个分区里面可以有好多天
      it =>{
        val jedis = JedisConnectionPool.getConnection()
        it.foreach(t=>{
          jedis.hincrByFloat(t._1,"total",t._2(0))
          jedis.hincrByFloat(t._1,"success",t._2(1))
          jedis.hincrByFloat(t._1,"money",t._2(2))
          jedis.hincrByFloat(t._1,"sumtime",t._2(3))
        })
        jedis.close()
        })
      //业务中每个小时的分布情况
      baseData.map(t => (t._1.substring(0,10),t._2))//按照日期来分
        .reduceByKey((list1,list2) =>{
        list1.zip(list2).map(t=>t._1+t._2)
      }).foreachPartition(
        //it 是RDD中每一个分区的数据，这个分区里面可以有好多天
        it =>{
          val jedis = JedisConnectionPool.getConnection()
          it.foreach(t=>{
            jedis.hincrByFloat(t._1,"total",t._2(0))
            jedis.hincrByFloat(t._1,"success",t._2(1))
            //设置key的有效期为一天
            jedis.expire(t._1,24*60*60)
          })
          jedis.close()
        })
      //2.计算每个省份充值成功的数量分布
      baseData.map(t => ((t._1.substring(0,8),t._3),t._2))//按照日期来分
        .reduceByKey((list1,list2) =>{
        list1.zip(list2).map(t=>t._1+t._2)
      }).foreachPartition(
        //it 是RDD中每一个分区的数据，这个分区里面可以有好多天
        it =>{
          val jedis = JedisConnectionPool.getConnection()
          val dict: Map[String, String] = pcodeBC.value
          it.foreach(t=>{
            //将每个省份的进行统计存入redis中的结果是
            // hashes key: PyyyyMMdd feild:[pcode] value:[成功的]
            //                       feild:[pcode] value:[成功的]
            //                       feild:[pcode] value:[成功的]
            //这里的到的省份是每个省份的code，想要的到省份的到省份的匹配规则中查找，需要写一个读取省份的方法
            jedis.hincrByFloat("p"+t._1._1,dict.getOrElse(t._1._2,t._1._2),t._2(1))

            //设置key的有效期为一天
            jedis.expire("p"+t._1,24*60*60)
          })
          jedis.close()
        })
      //3.统计每分钟的充值笔数及充值金额
      baseData.map(t => (t._1.substring(0,12),t._2)).reduceByKey{
        //这是偏函数的写法
        case(list1,list2)=>list1.zip(list2).map(t=>t._1+t._2)
      }.foreachPartition(
        //it 是RDD中每一个分区的数据，这个分区里面可以有好多天
        it =>{
          val jedis = JedisConnectionPool.getConnection()
          it.foreach(t=>{
            jedis.hincrByFloat("M"+t._1,"succ",t._2(1))
            jedis.hincrByFloat("M"+t._1,"money",t._2(2))
            //设置key的有效期为一天
            jedis.expire("M"+t._1,24*60*60)
          })
          jedis.close()
        })

      //将偏移量存入到mysql中
      OffsetUtils(offsetRanges,groupId)
    })





    //启动程序
    ssc.start()
    ssc.awaitTermination()


  }

}
